package com.huitongjy.common.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.net.*;
import java.text.DecimalFormat;
import java.util.*;
import java.util.concurrent.CountDownLatch;

/**
 * zookeeper连接客户端
 */
public class ZookeeperClient implements Watcher, AutoCloseable  {
    private ZooKeeper zookeeper;
    private static final int SESSION_TIME_OUT = 2000;
    private CountDownLatch countDownLatch = new CountDownLatch(1);


    public ZookeeperClient(String host) {
        try {
            zookeeper = new ZooKeeper(host, SESSION_TIME_OUT, this);
            countDownLatch.await();
        } catch (IOException | InterruptedException e) {
            System.out.println("zookeeper connect error");
            throw new RuntimeException(e);
        }
        System.out.println("zookeeper connect ok");
    }


    /**
     * 关闭zookeeper
     */
    @Override
    public void close(){
        try {
            zookeeper.close();
        } catch (InterruptedException e) {
            System.out.println("zookeeper close error");
        }
        zookeeper = null;
        System.out.println("zookeeper closed");
    }

    /**
     * 实现watcher的接口方法，当连接zookeeper成功后，zookeeper会通过此方法通知watcher
     * 此处为如果接受到连接成功的event，则countDown，让当前线程继续其他事情。
     */
    @Override
    public void process(WatchedEvent event) {
        if(event.getState() == Event.KeeperState.SyncConnected){
            System.out.println("watcher receiver event");
            countDownLatch.countDown();
        }
    }



    /**
     * 获取一个自增的机器Id (项目|host)唯一, 如果没有则生成
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    public Long getZookeeperWorkerId(String nodePath, String workerIdPath) {
        String node = nodePath + "/" + getLocalIP();
        try {
            mkdir(nodePath);

            String workerId;
            Stat exists = zookeeper.exists(node, false);

            if(exists==null) {
                //本机器 从来没有使用过
                workerId = initNewWorkerId(workerIdPath);
                if(workerId ==null) {
                    throw new RuntimeException("worker节点已满,请手动清除不用节点");
                }

                zookeeper.create(node, workerId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                System.out.println(node + "节点创建成功！");
            } else {
                //本机器 已经使用过
                byte[] data = zookeeper.getData(node, false, exists);
                workerId = new String(data);
            }
            return transLong(workerId);
        } catch (KeeperException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 生成一个新的workerId
     * @param workerIdPath workerId所在目录
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    private String initNewWorkerId(String workerIdPath) throws KeeperException, InterruptedException {
        mkdir(workerIdPath);

        String workerId = null;

        List<String> children = zookeeper.getChildren(workerIdPath, false);
        children.sort(Comparator.naturalOrder());
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            if(i == children.size() ||  i != Integer.valueOf(children.get(i))) {
                workerId = transStr(i);
                break;
            }
        }
        if(workerId!=null) {
            String result = zookeeper.create(workerIdPath + "/" + workerId, getCurrTimeMillis().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(result + "节点创建成功！");
        }
        return workerId;
    }



    /**
     * 创建目录
     * @param nodePath 创建目录
     * @throws KeeperException
     * @throws InterruptedException
     */
    private void mkdir(String nodePath) throws KeeperException, InterruptedException {
        Stat exists = zookeeper.exists(nodePath, false);
        if(exists != null) {
            return;
        }

        int index = nodePath.lastIndexOf("/");
        if(index>0) {
            mkdir(nodePath.substring(0, index));
        }
        zookeeper.create(nodePath,null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(nodePath + "节点创建成功！");
    }

    /**
     * 获取当前时间
     * @return 获取当前时间
     */
    private static String getCurrTimeMillis() {
        return String.valueOf(System.currentTimeMillis());
    }

    private static String transStr(long value) {
        return new DecimalFormat("0000000000").format(value);
    }

    private static Long transLong(String value) {
        return Long.valueOf(value);
    }

    /**
     * 本机的IP地址
     * @return 本机的IP地址
     */
    private static String getLocalIP(){
        Enumeration<NetworkInterface> allNetInterfaces;
        try {
            allNetInterfaces = NetworkInterface.getNetworkInterfaces();
        } catch (SocketException e) {
            throw new RuntimeException(e);
        }
        while (allNetInterfaces.hasMoreElements()){
            NetworkInterface netInterface = allNetInterfaces.nextElement();
            Enumeration<InetAddress> addresses = netInterface.getInetAddresses();
            while (addresses.hasMoreElements()){
                InetAddress ip= addresses.nextElement();
                //loopback地址即本机地址，IPv4的loopback范围是127.0.0.0 ~ 127.255.255.255
                if (ip != null && ip instanceof Inet4Address && !ip.isLoopbackAddress() && !ip.getHostAddress().contains(":")){
                    System.out.println("本机的IP = " + ip.getHostAddress());
                    return ip.getHostAddress();
                }
            }
        }
        throw new RuntimeException("获取本机Ip地址失败");
    }

    public static void main(String[] args) {
        System.out.println(getLocalIP());
    }
}
